
/*
 * 
 * Base class for managers
 */

package managers;

import java.util.Iterator;
import java.util.Random;
import java.util.LinkedList;
import java.util.ArrayList;

import core.Crawler;

import workers.BaseWorker;

import jobs.BaseJob;

public abstract class BaseManager extends Thread 
{	
	private LinkedList<BaseJob> mJobQueue;
	private ArrayList<BaseWorker> mWorkerList;
	
	private int mWorkerAvailability = 0;
	private Object mMonitor = new Object();
	private volatile boolean mRunning = true;
	private volatile boolean mAbort = false;
	
	private int mMaxQueueSize = 0;
	private int mMaxQueueSizeMultiplier = 5;
	
	public BaseManager( String name )
	{
		this.setName(name);
		mJobQueue = new LinkedList<BaseJob>();
		mWorkerList = new ArrayList<BaseWorker>();
	}
	
	/* Wakes up manager */
	public void notifyUpdate()
	{
		synchronized (mMonitor)
		{
			mMonitor.notify();
		}
	}
	
	/* Release one worker spot, called by workers to update when they are done */
	public void releaseLoad()
	{
		synchronized (mWorkerList)
		{
			mWorkerAvailability++;
		}
		notifyUpdate();
	}
	
	public void run()
	{
		while ( mRunning )
		{
			if ( isJobAvailable() )
			{
				if ( isWorkerAvailable() )
				{
					/* Assign Job from queue */
					assign(); 
				}
				
				if ( mAbort )
				{
					break;
				}		
				
				// wait to be notified by a worker
				synchronized ( mMonitor )
				{
					if ( !isWorkerAvailable() && mRunning )
					{
						try 
						{
							mMonitor.wait();
						} 
						catch (InterruptedException e) 
						{
							e.printStackTrace();
						}
					}
				}
			}
			
			// wait to be notified by a job
			synchronized ( mMonitor )
			{
				if ( !isJobAvailable() && mRunning)
				{
					try 
					{
						beforeGoingToSleepNoJob();
						mMonitor.wait();
					} 
					catch (InterruptedException e) 
					{
						e.printStackTrace();
					}
				}
			}
		}
		
		Iterator <BaseWorker> it = mWorkerList.iterator();
		while ( it.hasNext() )
		{
			try 
			{
				it.next().join();
			} 
			catch (InterruptedException e) 
			{
				e.printStackTrace();
			}
			}
	}
	
	/*
	 * find an available worker and push data to it
	 */
	private boolean assign()
	{		
		
		BaseJob job = null;
		BaseWorker worker = null;
		boolean found = false;	
		
		if ( !mRunning )
		{
			return false;
		}
		
		synchronized ( mJobQueue )
		{
			job = mJobQueue.remove();
		}
		
		synchronized ( mWorkerList )
		{		
			int count = 0;
			while ( count < mWorkerList.size() && mWorkerList.get( count ).hasJob() )
			{
				count++;
			}
			
			if ( count < mWorkerList.size() )
			{
				found = true;
				worker = mWorkerList.get( count );
				worker.setJob( job );
				mWorkerAvailability--;
			}
		}
		
		if ( found )
		{
			synchronized ( worker )
			{
				worker.notify();
			}
		}
		
		return true;
	}
	
	public int getWorkerCount()
	{
		synchronized (mWorkerList)
		{
			return mWorkerList.size();
		}
	}
	
	public int getAvailableWorkerCount()
	{
		synchronized ( mWorkerList )
		{
			return mWorkerAvailability;
		}
	}
	
	public int getBusyWorkerCount()
	{
		synchronized (mWorkerList)
		{
			return mWorkerList.size() - mWorkerAvailability;
		}
	}
	
	public int getQueueSize()
	{
		synchronized (mJobQueue)
		{
			return mJobQueue.size();
		}
	}
	
	/* Maximum number of jobs that can be inserted inside the queue */
	public int getQueueMaxSize()
	{
		return mMaxQueueSize;
	}
	
	public void setQueueSizeMultiplier( int size )
	{
		mMaxQueueSizeMultiplier = size;
	}
	
	/* Adds a job to the queue */
 	public boolean addJob( BaseJob job )
	{
 		if ( !mRunning )
		{
			return false;
		}
 		
 		boolean addProcess;
		synchronized ( mJobQueue )
		{
			/* Randomly replaces one of the jobs if the queue is full */
			if (mJobQueue.size() >= mMaxQueueSize )
			{
				Random generator = new Random();
				int roll = generator.nextInt(mJobQueue.size());
				boolean found = false;
				while ( !found )
				{
					if ( mJobQueue.get(roll).isImportant() )
					{
						roll = generator.nextInt(mJobQueue.size());
					}
					else
					{
						found = true;
					}
				}
				
				mJobQueue.add(roll, job);
				mJobQueue.remove(roll + 1);
				addProcess = true;
			}
			else
			{
				addProcess = mJobQueue.add( job );
			}
		}
		
		if ( addProcess )
		{
			notifyUpdate();
			return true;
		}
		return false;	
	}
	
 	/* remove job from queue */
	public boolean removeJob( BaseJob job )
	{
		if ( !mRunning )
		{
			return false;
		}
		
		synchronized ( mJobQueue )
		{
			return mJobQueue.remove( job );
		}
	}
	
	/* clear the job queue */
	public boolean removeAllJobs()
	{
		if ( !mRunning )
		{
			return false;
		}
		
		synchronized ( mJobQueue )
		{
			mJobQueue.clear();
		}
		
		return true;
	}
	
	/* Add a worker to the worker pool managed by this manager */
	public boolean addWorker( BaseWorker worker )
	{
		if ( !mRunning )
		{
			return false;
		}
		
		boolean addProcess;
		synchronized ( mWorkerList )
		{
			addProcess = mWorkerList.add( worker );
		}		
		
		if (addProcess)
		{
			mMaxQueueSize += mMaxQueueSizeMultiplier;
			worker.setManager( this );
			
			synchronized ( mWorkerList )
			{
				mWorkerAvailability++;
			}
			
			worker.start();
			return true;
		}
		
		return false;
	}
	
	/* Remove worker from the pool */
	public boolean removeWorker( BaseWorker worker )
	{
		if ( !mRunning )
		{
			return false;
		}
		
		boolean removeProcess;
		synchronized ( mWorkerList )
		{
			if ( worker.hasJob() )
			{
				return false;
			}
			
			removeProcess = mWorkerList.remove( worker );
		}
		
		if (removeProcess)
		{
			mMaxQueueSize -= mMaxQueueSizeMultiplier;
			
			synchronized ( mWorkerList )
			{
				mWorkerAvailability--;
			}
			
			return true;
		}
		
		return false;
	}
	
	/* Remove all workers from the pool */
	public boolean removeAllWorkers()
	{
		if ( !mRunning )
		{
			return false;
		}
		
		synchronized ( mWorkerList )
		{
			mWorkerList.clear();
		}
		
		return true;
	}
	
	/* one of the workers in it's pool doesn't have a job */
	public boolean isWorkerAvailable()
	{
	
		synchronized ( mWorkerList )
		{
			if ( mWorkerAvailability > 0 )
			{
				return true;
			}	
		}
		
		return false;
	}
	
	/* has job available in queue */
	public boolean isJobAvailable()
	{
		synchronized ( mJobQueue )
		{
			return !mJobQueue.isEmpty();
		}
	}
	
	public boolean isRunning()
	{
		return mRunning;
	}
	
	/* stop manager process */
	public void stopProcess()
	{			
		if ( !mRunning )
		{
			return;
		}
		
		stopWorkers();	
		
		synchronized ( mMonitor )
		{
			mRunning = false;
		}
		
		notifyUpdate();
	}
	
	/* stop workers */
	public void stopWorkers()
	{
		if ( !mRunning )
		{
			return;
		}
		
		synchronized ( mWorkerList )
		{
			Iterator<BaseWorker> iterator = mWorkerList.iterator();
			
			while ( iterator.hasNext() )
			{
				iterator.next().stopProcess();
			}
		}
	}
	
	/* abort manager and it's workers*/
	public void forcedStopProcess()
	{		
		if ( !mRunning )
		{
			return;
		}
		
		forcedStopWorkers();
		
		synchronized ( mMonitor )
		{
			mAbort = true;
			mRunning = false;
		}
	}
	
	public void forcedStopWorkers()
	{
		if ( !mRunning )
		{
			return;
		}
		
		synchronized ( mWorkerList )
		{
			Iterator<BaseWorker> iterator = mWorkerList.iterator();
			
			while ( iterator.hasNext() )
			{
				iterator.next().forcedStopProcess();
			}
		}
	}
	
	protected void beforeGoingToSleepNoJob()
	{
		
	}
}